package com.yusc.socket;

import com.google.gson.Gson;
import com.yusc.pojo.ALlBasicInfo;
import com.yusc.pojo.Integral;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Random;

/**
 * @Author: Yusc
 * @Date: 2023/10/23 14:35
 * @Version:
 * @Description: 服务器端 --> 模拟随机一名会员进行了一次积分兑换
 **/
@Slf4j
public class Server {
    private static Gson gson = new Gson();

    public static void main(String[] args) throws IOException, InterruptedException {
        //创建服务器套接字并绑定接口
        ServerSocket serverSocket = new ServerSocket(9999);
        System.out.println("服务端已经启动，等待客户机连接...");

        //监听客户段连接请求
        Socket clientServer = serverSocket.accept();
        System.out.println("客户端已经连接，IP地址:" + clientServer.getInetAddress());

        //创建输入流和输出流
        BufferedReader in = new BufferedReader(new InputStreamReader(clientServer.getInputStream()));
        PrintWriter out = new PrintWriter(clientServer.getOutputStream(), true);


        SparkConf sparkConf = new SparkConf().setAppName("Points redemption times").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        JavaRDD<String> rdd = sc.textFile("src/main/resources/data/test.json");
        Random r = new Random();

        List<Integral> collect = rdd.map(x -> {
                    // 将所需字段封装到对象中
                    ALlBasicInfo allInfo = gson.fromJson(x, ALlBasicInfo.class);
                    return new Integral(allInfo.getMemberNo(), allInfo.getFfpTier(),
                            allInfo.getWorkCity(), allInfo.getAge(), null, null);
                })
                .collect();

        int index = 0;
        do {
            Integral integral = collect.get(r.nextInt(collect.size()));
            integral.setExchangeTime(LocalDateTime.now());
            String jsonInfo = gson.toJson(integral);  // 随机一位会员兑换
            log.info("{}", jsonInfo);
            out.println(jsonInfo);
            index++;
            Thread.sleep(r.nextInt(1000) * 5 + 1);  // 停0到5秒 方便观察
        } while (index != 100);

        // 关闭连接
        in.close();
        out.close();
        clientServer.close();
        serverSocket.close();
        sc.stop();
    }
}
